package com.learn.kafka;

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.PartitionInfo;

import java.util.List;
import java.util.Map;
import java.util.Random;

/**
 * 自定义分区
 *
 * @author: MI
 * @email: 448341911@qq.com
 * @createTime: 2025/8/5 15:05
 * @updateUser: MI
 * @updateTime: 2025/8/5 15:05
 * @updateRemark: 修改内容
 * @version: 1.0
 */
public class MyPartition implements Partitioner {

    private Random random = new Random();

    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        System.out.println("enter");
        // 获取所有topic 分区
        List<PartitionInfo> list = cluster.partitionsForTopic(topic);
        int partitionNum = 0;
        int len = list.size();
        if (null == key) {
            // 随机分区
            partitionNum = random.nextInt(len);
        } else {
            partitionNum = Math.abs(key.hashCode() % len);
        }
        System.out.println("key ->" + key + "->value->" + value + "->" + partitionNum);
        return partitionNum; // 指定发送的 分区值
    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> configs) {

    }
}
